package day04.actorPlus

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
import scala.collection.mutable

/**
  *
  */
class Master extends Actor{

  //启动一个定时器，定期检查超时的worker
  @scala.throws[Exception](classOf[Exception])
  override def preStart(): Unit = {
    import context.dispatcher
    context.system.scheduler.schedule(0 millis,15000 millis,self,CheckTimeOutWorker)
  }

  //用来保存注册信息
  val map = new mutable.HashMap[String,WorkerInfo]()


  override def receive: Receive = {
    //接受到worker的注册消息
    case RegisterWorker(workerId,memory,cores) => {
      println("有worker过来注册了"+workerId)
      //封装workerInfo 信息，然后将worker信息保存起来（map）
      val workerInfo = new WorkerInfo(workerId,memory,cores)
      if(!map.contains(workerId)){
        map.put(workerId,workerInfo)
        //master给worker返回注册成功的消息
        sender() ! RegisteredWorker
      }

    }
    case HeartBeat(workerId) =>{
      //更新上次心跳时间
      val current = System.currentTimeMillis()
      //根据workerid 到map中找到相对应的 woekerInfo
      val workerInfo = map(workerId)
      workerInfo.lastHeartbeanTime = current
    }

    case CheckTimeOutWorker=>{
      val workers: Iterable[WorkerInfo] = map.values
      //检查哪些worker 超时了，取出所有的进行当前时间做对比
      val current = System.currentTimeMillis()

      val deadWorkers: Iterable[WorkerInfo] = workers.filter(workerInfo => {
        current - workerInfo.lastHeartbeanTime > 20000
      })

      //从map中移除超时的worker
      for(dw <- deadWorkers){
        map-=dw.workerId
      }

      //deadWorkers.foreach(dw => {map -= dw.workerId})

      println("current live worker size is"+map.size)

    }
  }
}

object Master{
  //在Master的object中定义一个常量
  val MASTER_SYSTEM = "Master_System"
  val MASTER_NAME = "Master"


  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port ="8888"

    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = $host
         |akka.remote.netty.tcp.port = $port
      """.stripMargin

    val config = ConfigFactory.parseString(confStr)
    // 创建 actorSystem(单例)
    val actorSystem = ActorSystem(MASTER_SYSTEM,config)

    //通过创建的actorSystem 在创建actor
    val masterActor: ActorRef = actorSystem.actorOf(Props[Master](),MASTER_NAME)


    //发送消息（给自己发）
    masterActor.!("hello")



  }
}

